package com.cjfather.redis.steam.handler;

import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.ErrorHandler;

import io.fury.util.StringUtils;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;

/**
 * StreamPollTask 获取消息或对应的listener消费消息过程中发生了异常
 *
 * @author todd 2023-07-17 - 下午3:44
 */
@Slf4j
public class CustomErrorHandler implements ErrorHandler {

	private RedisTemplate template;
	
	private ReactiveRedisTemplate rtemplate;
	
	private String key;
	
	private String group;
	
	private String cname;
	
	public CustomErrorHandler(RedisTemplate template, String key, String group,String cname) {
		this.template = template;
		this.key = key;
		this.group = group;
		this.cname = cname;
	}
	
	public CustomErrorHandler(ReactiveRedisTemplate rtemplate, String key, String group,String cname) {
		this.rtemplate = rtemplate;
		this.key = key;
		this.group = group;
		this.cname = cname;
	}

    @Override
    public void handleError(Throwable t) {
    	try {
			if(t.getMessage().contains("NOGROUP No such key")) {
				if(template != null){
					template.opsForStream().createGroup(key,ReadOffset.from("0-0"), group);
				}else {
					rtemplate.opsForStream().createGroup(key,ReadOffset.from("0-0"), group);
				}
				log.info("create {} key on group {}", key, group);
			}
		} catch (Exception e) {
			log.error("handler err:" + e.getMessage(),e);
		}
        log.error("{} stream err:" + t.getMessage(),cname, t);
    }
    
    public static void main(String[] args) {
    	Mono<Boolean> mono = Mono.just(true);
    	System.out.println(mono.block());
    	
	}
}
